package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

import java.sql.{Connection, DriverManager, PreparedStatement}

object ForeachPartitionDemo {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core:        SparkContext -> RDD
     * Spark Sql:         SparkSession -> DataFrame
     * Spark Streaming:   StreamingContext -> DStream
     */
    //this(sparkContext: SparkContext, batchDuration: Duration)

    val ss: SparkSession = SparkSession.builder()
      .master("local[*]") // 如果不给线程核数，默认只有1个 // 放在yarn中的话，不需要local
      .appName("word count")
      //      .config("spark.default.parallelism", "1")
      //      .config("spark.streaming.concurrentJobs",1)
      .getOrCreate()

    import ss.implicits._
    import org.apache.spark.sql.functions._

    val sc: SparkContext = ss.sparkContext

    //    val conf = new SparkConf()
    //    conf.setMaster("local")
    //    conf.setAppName("word count")
    //    val sc = new SparkContext(conf)

    //传入一个SparkContext对象，以及接收要处理的数据时间段
    val streamingContext = new StreamingContext(sc, Durations.seconds(5L))

    //近实时处理中，数据应当是源源不断的
    //使用端口号的数据，模拟源源不断的数据源
    //监控对应端口号中的数据
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    //    /**
    //     * foreachRDD函数没有返回值，处理完rdd数据就结束，不会返回新的DStream对象
    //     * transform函数也可以单独对DStream中RDD进行处理，但是与foreachRDD不同点在于，transform函数可以返回一个新的DStream对象
    //     */
    //    infoDS.foreachRDD((rdd: RDD[String]) => {
    //      println("====================正在处理rdd====================")
    //
    //      /**
    //       * 纯RDD操作，计算单词统计
    //       * 1500100001,施笑槐,22,女,文科六班
    //       */
    //      rdd.repartition(1).foreach((line: String) => {
    //        println("--------------建立数据库连接----------------------")
    //        //构建数据库连接对象
    //        Class.forName("com.mysql.jdbc.Driver")
    //        val conn: Connection = DriverManager.getConnection(
    //          "jdbc:mysql://master:3306/bigdata33?useUnicode=true&characterEncoding=utf-8&useSSL=false",
    //          "root",
    //          "123456"
    //        )
    //
    //        val prep: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")
    //
    //        val infos: Array[String] = line.split(",")
    //        val id: String = infos(0)
    //        val name: String = infos(1)
    //        val age: String = infos(2)
    //        val gender: String = infos(3)
    //        val clazz: String = infos(4)
    //
    //        prep.setString(1, id)
    //        prep.setString(2, name)
    //        prep.setString(3, age)
    //        prep.setString(4, gender)
    //        prep.setString(5, clazz)
    //
    //        prep.executeUpdate()
    //
    //        //释放资源
    //        prep.close()
    //        conn.close()
    //
    //      })
    //    })


    /**
     * 上面的rdd处理过程中，每条数据都会建立一次连接，非常消耗资源
     */
    //    infoDS.foreachRDD((rdd: RDD[String]) => {
    //      println("====================正在处理rdd====================")
    //      println("--------------建立数据库连接----------------------")
    //      //构建数据库连接对象
    //      Class.forName("com.mysql.jdbc.Driver")
    //      val conn: Connection = DriverManager.getConnection(
    //        "jdbc:mysql://master:3306/bigdata33?useUnicode=true&characterEncoding=utf-8&useSSL=false",
    //        "root",
    //        "123456"
    //      )
    //
    //      val prep: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")
    //
    //      /**
    //       * 纯RDD操作，计算单词统计
    //       * 1500100001,施笑槐,22,女,文科六班
    //       */
    //      rdd.repartition(1).foreach((line: String) => {
    //        val infos: Array[String] = line.split(",")
    //        val id: String = infos(0)
    //        val name: String = infos(1)
    //        val age: String = infos(2)
    //        val gender: String = infos(3)
    //        val clazz: String = infos(4)
    //
    //        prep.setString(1, id)
    //        prep.setString(2, name)
    //        prep.setString(3, age)
    //        prep.setString(4, gender)
    //        prep.setString(5, clazz)
    //
    //        prep.executeUpdate()
    //      })
    //
    //      //释放资源
    //      prep.close()
    //      conn.close()
    //    })


    /**
     * sparkstream中对DStream提供一个新的函数，可以对每一个rdd中分区进行处理
     */


    infoDS.foreachRDD((rdd: RDD[String]) => {
      println("====================正在处理rdd====================")

      //对每一个rdd中的分区单独处理
      val rdd2: RDD[String] = rdd.repartition(2)
      println(s"====================rdd共有${rdd2.getNumPartitions} 分区====================")
      rdd2.foreachPartition((itr: Iterator[String]) => {
        println("--------------建立数据库连接----------------------")
        //构建数据库连接对象
        Class.forName("com.mysql.jdbc.Driver")
        val conn: Connection = DriverManager.getConnection(
          "jdbc:mysql://master:3306/bigdata33?useUnicode=true&characterEncoding=utf-8&useSSL=false",
          "root",
          "123456"
        )

        val prep: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")

        itr.foreach((line: String) => {
          //处理分区中的每条数据
          val infos: Array[String] = line.split(",")
          val id: String = infos(0)
          val name: String = infos(1)
          val age: String = infos(2)
          val gender: String = infos(3)
          val clazz: String = infos(4)

          prep.setString(1, id)
          prep.setString(2, name)
          prep.setString(3, age)
          prep.setString(4, gender)
          prep.setString(5, clazz)

          prep.executeUpdate()
        })

        //释放资源
        prep.close()
        conn.close()
      })


    })


    /**
     * 额外提交命令，一直启动sparkstreaming程序
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }

}
